<!DOCTYPE html>
<html lang="zh">
<head>
  <meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1, maximum-scale=2">
<meta name="theme-color" content="#222">
<meta name="generator" content="Hexo 5.3.0">


  <link rel="apple-touch-icon" sizes="180x180" href="/yuwanzi.io/images/apple-touch-icon-next.png">
  <link rel="icon" type="image/png" sizes="32x32" href="/yuwanzi.io/images/favicon-32x32-next.png">
  <link rel="icon" type="image/png" sizes="16x16" href="/yuwanzi.io/images/favicon-16x16-next.png">
  <link rel="mask-icon" href="/yuwanzi.io/images/logo.svg" color="#222">

<link rel="stylesheet" href="/yuwanzi.io/css/main.css">



<link rel="stylesheet" href="//cdn.jsdelivr.net/npm/@fortawesome/fontawesome-free@5.15.1/css/all.min.css">
  <link rel="stylesheet" href="//cdn.jsdelivr.net/npm/animate.css@3.1.1/animate.min.css">

<script class="hexo-configurations">
    var NexT = window.NexT || {};
    var CONFIG = {"hostname":"suyuhuan.gitee.io","root":"/yuwanzi.io/","images":"/yuwanzi.io/images","scheme":"Muse","version":"8.2.0","exturl":false,"sidebar":{"position":"left","display":"post","padding":18,"offset":12},"copycode":false,"bookmark":{"enable":false,"color":"#222","save":"auto"},"fancybox":false,"mediumzoom":false,"lazyload":false,"pangu":false,"comments":{"style":"tabs","active":null,"storage":true,"lazyload":false,"nav":null},"motion":{"enable":true,"async":false,"transition":{"post_block":"fadeIn","post_header":"fadeInDown","post_body":"fadeInDown","coll_header":"fadeInLeft","sidebar":"fadeInUp"}},"prism":false,"i18n":{"placeholder":"Suche...","empty":"We didn't find any results for the search: ${query}","hits_time":"${hits} results found in ${time} ms","hits":"${hits} results found"}};
  </script>
<meta name="description" content="什么是MapReduce&amp;nbsp;&amp;nbsp;MapReduce是一种分布式计算模型，由Google提出，主要用于搜索领域，解决海量数据的计算问题。 &amp;nbsp;&amp;nbsp;MapReduce是处理大量半结构化数据集合的编程模型。编程模型是一种处理并结构化特定问题的方式。例如，在一个关系数据库中，使用一种集合语言执行查询，如SQL。告诉语言想要的结果，并将它提交给系统来计算出如何产生计算。还可">
<meta property="og:type" content="article">
<meta property="og:title" content="Hadoop学习笔记(2)-Mapreduce">
<meta property="og:url" content="https://suyuhuan.gitee.io/yuwanzi.io/2016/07/14/2016-07-14-Hadoop02-MapReduce/index.html">
<meta property="og:site_name" content="玉丸子 | Blog">
<meta property="og:description" content="什么是MapReduce&amp;nbsp;&amp;nbsp;MapReduce是一种分布式计算模型，由Google提出，主要用于搜索领域，解决海量数据的计算问题。 &amp;nbsp;&amp;nbsp;MapReduce是处理大量半结构化数据集合的编程模型。编程模型是一种处理并结构化特定问题的方式。例如，在一个关系数据库中，使用一种集合语言执行查询，如SQL。告诉语言想要的结果，并将它提交给系统来计算出如何产生计算。还可">
<meta property="og:locale">
<meta property="og:image" content="http://ww4.sinaimg.cn/mw690/63503acbjw1f5w446w8pcj20bp08rweu.jpg">
<meta property="og:image" content="http://ww4.sinaimg.cn/mw690/63503acbjw1f5w4495s9rj20sa0foac1.jpg">
<meta property="og:image" content="http://ww3.sinaimg.cn/mw690/63503acbjw1f5w44b7lxfj208u05dt9v.jpg">
<meta property="og:image" content="http://ww1.sinaimg.cn/mw690/63503acbjw1f5w44bjrc6j206m05jq3d.jpg">
<meta property="og:image" content="http://ww1.sinaimg.cn/mw690/63503acbjw1f5w48pm9byj20ui0kl0uy.jpg">
<meta property="og:image" content="http://ww3.sinaimg.cn/mw690/63503acbjw1f5w449wf5hj20mv0aodh4.jpg">
<meta property="og:image" content="http://ww4.sinaimg.cn/mw690/63503acbjw1f5w44an439j20k10g5mzg.jpg">
<meta property="article:published_time" content="2016-07-14T10:00:00.000Z">
<meta property="article:modified_time" content="2020-11-07T00:58:17.000Z">
<meta property="article:author" content="玉丸子">
<meta property="article:tag" content="Hadoop">
<meta property="article:tag" content="大数据">
<meta name="twitter:card" content="summary">
<meta name="twitter:image" content="http://ww4.sinaimg.cn/mw690/63503acbjw1f5w446w8pcj20bp08rweu.jpg">


<link rel="canonical" href="https://suyuhuan.gitee.io/yuwanzi.io/2016/07/14/2016-07-14-Hadoop02-MapReduce/">


<script class="page-configurations">
  // https://hexo.io/docs/variables.html
  CONFIG.page = {
    sidebar: "",
    isHome : false,
    isPost : true,
    lang   : 'zh'
  };
</script>
<title>Hadoop学习笔记(2)-Mapreduce | 玉丸子 | Blog</title>
  




  <noscript>
  <style>
  body { margin-top: 2rem; }

  .use-motion .menu-item,
  .use-motion .sidebar,
  .use-motion .post-block,
  .use-motion .pagination,
  .use-motion .comments,
  .use-motion .post-header,
  .use-motion .post-body,
  .use-motion .collection-header {
    visibility: visible;
  }

  .use-motion .header,
  .use-motion .site-brand-container .toggle,
  .use-motion .footer { opacity: initial; }

  .use-motion .site-title,
  .use-motion .site-subtitle,
  .use-motion .custom-logo-image {
    opacity: initial;
    top: initial;
  }

  .use-motion .logo-line {
    transform: scaleX(1);
  }

  .search-pop-overlay, .sidebar-nav { display: none; }
  .sidebar-panel { display: block; }
  </style>
</noscript>

<link rel="alternate" href="/yuwanzi.io/atom.xml" title="玉丸子 | Blog" type="application/atom+xml">
</head>

<body itemscope itemtype="http://schema.org/WebPage" class="use-motion">
  <div class="headband"></div>

  <main class="main">
    <header class="header" itemscope itemtype="http://schema.org/WPHeader">
      <div class="header-inner"><div class="site-brand-container">
  <div class="site-nav-toggle">
    <div class="toggle" aria-label="Navigationsleiste an/ausschalten" role="button">
    </div>
  </div>

  <div class="site-meta">

    <a href="/yuwanzi.io/" class="brand" rel="start">
      <i class="logo-line"></i>
      <h1 class="site-title">玉丸子 | Blog</h1>
      <i class="logo-line"></i>
    </a>
  </div>

  <div class="site-nav-right">
    <div class="toggle popup-trigger">
    </div>
  </div>
</div>







</div>
        
  
  <div class="toggle sidebar-toggle" role="button">
    <span class="toggle-line"></span>
    <span class="toggle-line"></span>
    <span class="toggle-line"></span>
  </div>

  <aside class="sidebar">

    <div class="sidebar-inner sidebar-nav-active sidebar-toc-active">
      <ul class="sidebar-nav">
        <li class="sidebar-nav-toc">
          Inhaltsverzeichnis
        </li>
        <li class="sidebar-nav-overview">
          Übersicht
        </li>
      </ul>

      <div class="sidebar-panel-container">
        <!--noindex-->
        <div class="post-toc-wrap sidebar-panel">
            <div class="post-toc animated"><ol class="nav"><li class="nav-item nav-level-3"><a class="nav-link" href="#%E4%BB%80%E4%B9%88%E6%98%AFMapReduce"><span class="nav-number">1.</span> <span class="nav-text">什么是MapReduce</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#Yarn%E6%A6%82%E8%BF%B0"><span class="nav-number">2.</span> <span class="nav-text">Yarn概述</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#%E5%8E%9FMapReduce%E6%9E%B6%E6%9E%84%E7%9A%84%E4%B8%8D%E8%B6%B3"><span class="nav-number">3.</span> <span class="nav-text">原MapReduce架构的不足</span></a></li><li class="nav-item nav-level-3"><a class="nav-link" href="#MRv2-Yarn%E5%B7%A5%E4%BD%9C%E6%B5%81%E7%A8%8B"><span class="nav-number">4.</span> <span class="nav-text">MRv2&#x2F;Yarn工作流程</span></a><ol class="nav-child"><li class="nav-item nav-level-4"><a class="nav-link" href="#Yarn%E6%9E%B6%E6%9E%84"><span class="nav-number">4.1.</span> <span class="nav-text">Yarn架构</span></a></li><li class="nav-item nav-level-4"><a class="nav-link" href="#%E5%B7%A5%E4%BD%9C%E6%B5%81%E7%A8%8B"><span class="nav-number">4.2.</span> <span class="nav-text">工作流程</span></a></li></ol></li><li class="nav-item nav-level-3"><a class="nav-link" href="#Shuffle%E8%BF%87%E7%A8%8B"><span class="nav-number">5.</span> <span class="nav-text">Shuffle过程</span></a><ol class="nav-child"><li class="nav-item nav-level-4"><a class="nav-link" href="#Map"><span class="nav-number">5.1.</span> <span class="nav-text">Map</span></a></li><li class="nav-item nav-level-4"><a class="nav-link" href="#Reduce"><span class="nav-number">5.2.</span> <span class="nav-text">Reduce</span></a></li></ol></li><li class="nav-item nav-level-3"><a class="nav-link" href="#WordCount%E6%A1%88%E4%BE%8B"><span class="nav-number">6.</span> <span class="nav-text">WordCount案例</span></a><ol class="nav-child"><li class="nav-item nav-level-4"><a class="nav-link" href="#Mapper"><span class="nav-number">6.1.</span> <span class="nav-text">Mapper</span></a></li><li class="nav-item nav-level-4"><a class="nav-link" href="#Reducer"><span class="nav-number">6.2.</span> <span class="nav-text">Reducer</span></a></li><li class="nav-item nav-level-4"><a class="nav-link" href="#Main"><span class="nav-number">6.3.</span> <span class="nav-text">Main</span></a></li><li class="nav-item nav-level-4"><a class="nav-link" href="#%E5%90%AF%E5%8A%A8MapReduce"><span class="nav-number">6.4.</span> <span class="nav-text">启动MapReduce</span></a></li></ol></li><li class="nav-item nav-level-3"><a class="nav-link" href="#%E8%87%AA%E5%AE%9A%E4%B9%89Sort"><span class="nav-number">7.</span> <span class="nav-text">自定义Sort</span></a><ol class="nav-child"><li class="nav-item nav-level-4"><a class="nav-link" href="#bean"><span class="nav-number">7.1.</span> <span class="nav-text">bean</span></a></li><li class="nav-item nav-level-4"><a class="nav-link" href="#main"><span class="nav-number">7.2.</span> <span class="nav-text">main</span></a></li></ol></li><li class="nav-item nav-level-3"><a class="nav-link" href="#%E8%87%AA%E5%AE%9A%E4%B9%89Partition"><span class="nav-number">8.</span> <span class="nav-text">自定义Partition</span></a><ol class="nav-child"><li class="nav-item nav-level-4"><a class="nav-link" href="#partitioner"><span class="nav-number">8.1.</span> <span class="nav-text">partitioner</span></a></li><li class="nav-item nav-level-4"><a class="nav-link" href="#main-1"><span class="nav-number">8.2.</span> <span class="nav-text">main</span></a></li></ol></li><li class="nav-item nav-level-3"><a class="nav-link" href="#END"><span class="nav-number">9.</span> <span class="nav-text">END</span></a></li></ol></div>
        </div>
        <!--/noindex-->

        <div class="site-overview-wrap sidebar-panel">
          <div class="site-author site-overview-item animated" itemprop="author" itemscope itemtype="http://schema.org/Person">
  <p class="site-author-name" itemprop="name">玉丸子</p>
  <div class="site-description" itemprop="description">这里是玉丸子的个人博客,与你一起发现更大的世界。</div>
</div>
<div class="site-state-wrap site-overview-item animated">
  <nav class="site-state">
      <div class="site-state-item site-state-posts">
          <a href="/yuwanzi.io/archives">
          <span class="site-state-item-count">68</span>
          <span class="site-state-item-name">Artikel</span>
        </a>
      </div>
      <div class="site-state-item site-state-categories">
            <a href="/yuwanzi.io/categories/">
        <span class="site-state-item-count">39</span>
        <span class="site-state-item-name">Kategorien</span></a>
      </div>
      <div class="site-state-item site-state-tags">
            <a href="/yuwanzi.io/tags/">
        <span class="site-state-item-count">46</span>
        <span class="site-state-item-name">schlagwörter</span></a>
      </div>
  </nav>
</div>



        </div>
      </div>
    </div>
  </aside>
  <div class="sidebar-dimmer"></div>


    </header>

    
  <div class="back-to-top" role="button">
    <i class="fa fa-arrow-up"></i>
    <span>0%</span>
  </div>

<noscript>
  <div class="noscript-warning">Theme NexT works best with JavaScript enabled</div>
</noscript>


    <div class="main-inner post posts-expand">


  


<div class="post-block">
  
  

  <article itemscope itemtype="http://schema.org/Article" class="post-content" lang="zh">
    <link itemprop="mainEntityOfPage" href="https://suyuhuan.gitee.io/yuwanzi.io/2016/07/14/2016-07-14-Hadoop02-MapReduce/">

    <span hidden itemprop="author" itemscope itemtype="http://schema.org/Person">
      <meta itemprop="image" content="/yuwanzi.io/images/avatar.gif">
      <meta itemprop="name" content="玉丸子">
      <meta itemprop="description" content="这里是玉丸子的个人博客,与你一起发现更大的世界。">
    </span>

    <span hidden itemprop="publisher" itemscope itemtype="http://schema.org/Organization">
      <meta itemprop="name" content="玉丸子 | Blog">
    </span>
      <header class="post-header">
        <h1 class="post-title" itemprop="name headline">
          Hadoop学习笔记(2)-Mapreduce
        </h1>

        <div class="post-meta-container">
          <div class="post-meta">
    <span class="post-meta-item">
      <span class="post-meta-item-icon">
        <i class="far fa-calendar"></i>
      </span>
      <span class="post-meta-item-text">Veröffentlicht am</span>

      <time title="Erstellt: 2016-07-14 18:00:00" itemprop="dateCreated datePublished" datetime="2016-07-14T18:00:00+08:00">2016-07-14</time>
    </span>
      <span class="post-meta-item">
        <span class="post-meta-item-icon">
          <i class="far fa-calendar-check"></i>
        </span>
        <span class="post-meta-item-text">Bearbeitet am</span>
        <time title="Geändert am: 2020-11-07 08:58:17" itemprop="dateModified" datetime="2020-11-07T08:58:17+08:00">2020-11-07</time>
      </span>
    <span class="post-meta-item">
      <span class="post-meta-item-icon">
        <i class="far fa-folder"></i>
      </span>
      <span class="post-meta-item-text">in</span>
        <span itemprop="about" itemscope itemtype="http://schema.org/Thing">
          <a href="/yuwanzi.io/categories/%E5%90%8E%E7%AB%AF/" itemprop="url" rel="index"><span itemprop="name">后端</span></a>
        </span>
          . 
        <span itemprop="about" itemscope itemtype="http://schema.org/Thing">
          <a href="/yuwanzi.io/categories/%E5%90%8E%E7%AB%AF/%E5%A4%A7%E6%95%B0%E6%8D%AE/" itemprop="url" rel="index"><span itemprop="name">大数据</span></a>
        </span>
          . 
        <span itemprop="about" itemscope itemtype="http://schema.org/Thing">
          <a href="/yuwanzi.io/categories/%E5%90%8E%E7%AB%AF/%E5%A4%A7%E6%95%B0%E6%8D%AE/Hadoop/" itemprop="url" rel="index"><span itemprop="name">Hadoop</span></a>
        </span>
    </span>

  
</div>

        </div>
      </header>

    
    
    
    <div class="post-body" itemprop="articleBody">
        <p><img src="http://ww4.sinaimg.cn/mw690/63503acbjw1f5w446w8pcj20bp08rweu.jpg"></p>
<h3 id="什么是MapReduce"><a href="#什么是MapReduce" class="headerlink" title="什么是MapReduce"></a>什么是MapReduce</h3><p>&nbsp;&nbsp;MapReduce是一种分布式计算模型，由Google提出，主要用于搜索领域，解决海量数据的计算问题。</p>
<p>&nbsp;&nbsp;MapReduce是处理大量半结构化数据集合的编程模型。编程模型是一种处理并结构化特定问题的方式。例如，在一个关系数据库中，使用一种集合语言执行查询，如SQL。告诉语言想要的结果，并将它提交给系统来计算出如何产生计算。还可以用更传统的语言(C++，Java)，一步步地来解决问题。这是两种不同的编程模型，MapReduce就是另外一种。</p>
<p>&nbsp;&nbsp;MapReduce和Hadoop是相互独立的，实际上又能相互配合工作得很好。</p>
<h3 id="Yarn概述"><a href="#Yarn概述" class="headerlink" title="Yarn概述"></a>Yarn概述</h3><p>&nbsp;&nbsp;Yarn是一个分布式的资源管理系统，用以提高分布式的集群环境下的资源利用率，这些资源包括内存、IO、网络、磁盘等。其产生的原因是为了解决原MapReduce框架的不足。最初MapReduce的committer们还可以周期性的在已有的代码上进行修改，可是随着代码的增加以及原MapReduce框架设计的不足，在原MapReduce框架上进行修改变得越来越困难，所以MapReduce的committer们决定从架构上重新设计MapReduce,使下一代的MapReduce(MRv2/Yarn)框架具有更好的扩展性、可用性、可靠性、向后兼容性和更高的资源利用率以及能支持除了MapReduce计算框架外的更多的计算框架。</p>
<h3 id="原MapReduce架构的不足"><a href="#原MapReduce架构的不足" class="headerlink" title="原MapReduce架构的不足"></a>原MapReduce架构的不足</h3><p><img src="http://ww4.sinaimg.cn/mw690/63503acbjw1f5w4495s9rj20sa0foac1.jpg"></p>
<ul>
<li>JobTracker是集群事务的集中处理点，存在单点故障。</li>
<li>JobTracker需要完成的任务太多，既要维护job的状态又要维护job的task的状态，造成过多的资源消耗。</li>
<li>在taskTracker端，用map/reduce task作为资源的表示过于简单，没有考虑到CPU、内存等资源情况，当把两个需要消耗大内存的task调度到一起，很容易出现OOM(Out Of Memory内存不足)。</li>
<li>把资源强制划分为map/reduce slot,当只有map task时，reduce slot不能用；当只有reduce task时，map slot不能用，容易造成资源利用不足。</li>
</ul>
<h3 id="MRv2-Yarn工作流程"><a href="#MRv2-Yarn工作流程" class="headerlink" title="MRv2/Yarn工作流程"></a>MRv2/Yarn工作流程</h3><h4 id="Yarn架构"><a href="#Yarn架构" class="headerlink" title="Yarn架构"></a>Yarn架构</h4><p>&nbsp;&nbsp;Yarn/MRv2最基本的想法是将原JobTracker主要的资源管理和job调度/监视功能分开作为两个单独的守护进程。</p>
<p>&nbsp;&nbsp;有一个全局的ResourceManager(RM)和每个Application有一个ApplicationMaster(AM)，Application相当于map-reduce job或者DAG jobs。</p>
<p>&nbsp;&nbsp;ResourceManager和NodeManager(NM)组成了基本的数据计算框架。ResourceManager协调集群的资源利用，任何client或者运行着的applicatitonMaster想要运行job或者task都得向RM申请一定的资源。ApplicatonMaster是一个框架特殊的库，对于MapReduce框架而言有它自己的AM实现，用户也可以实现自己的AM，在运行的时候，AM会与NM一起来启动和监视tasks。 </p>
<p><strong>ResourceManager</strong></p>
<p>ResourceManager作为资源的协调者有两个主要的组件：Scheduler和ApplicationsManager(AsM)。</p>
<p>Scheduler负责分配最少但满足application运行所需的资源量给Application。Scheduler只是基于资源的使用情况进行调度，并不负责监视/跟踪application的状态，当然也不会处理失败的task。RM使用resource container概念来管理集群的资源，resource container是资源的抽象，每个container包括一定的内存、IO、网络等资源，不过目前的实现只包括内存一种资源。</p>
<p>ApplicationsManager负责处理client提交的job以及协商第一个container以供applicationMaster运行，并且在applicationMaster失败的时候会重新启动applicationMaster。下面阐述RM具体完成的一些功能。</p>
<ol>
<li><p>资源调度：Scheduler从所有运行着的application收到资源请求后构建一个全局的资源分配计划，然后根据application特殊的限制以及全局的一些限制条件分配资源。</p>
</li>
<li><p>资源监视：Scheduler会周期性的接收来自NM的资源使用率的监控信息，另外applicationMaster可以从Scheduler得到属于它的已完成的container的状态信息。</p>
</li>
<li><p>Application提交：</p>
<ul>
<li>client向AsM获得一个applicationIDclient将application定义以及需要的jar包.</li>
<li>client将application定义以及需要的jar包文件等上传到hdfs的指定目录，由yarn-site.xml的yarn.app.mapreduce.am.staging-dir指定.</li>
<li>client构造资源请求的对象以及application的提交context发送给AsM.</li>
<li>AsM接收application的提交context.</li>
<li>AsM根据application的信息向Scheduler协商一个Container供applicationMaster运行，然后启动applicationMaster.</li>
<li>向该container所属的NM发送launchContainer信息启动该container,也即启动applicationMaster、AsM向client提供运行着的AM的状态信息.</li>
</ul>
</li>
<li><p> AM的生命周期：AsM负责系统中所有AM的生命周期的管理。AsM负责AM的启动，当AM启动后，AM会周期性的向AsM发送heartbeat，默认是1s，AsM据此了解AM的存活情况，并且在AM fail时负责重启AM，若是一定时间过后(默认10分钟)没有收到AM的heartbeat，AsM就认为该AM已经fail。</p>
</li>
</ol>
<p><strong>NodeManager</strong></p>
<p>&nbsp;&nbsp;NM主要负责启动RM分配给AM的container以及代表AM的container，并且会监视container的运行情况。在启动container的时候，NM会设置一些必要的环境变量以及将container运行所需的jar包、文件等从hdfs下载到本地，也就是所谓的资源本地化；当所有准备工作做好后，才会启动代表该container的脚本将程序启动起来。启动起来后，NM会周期性的监视该container运行占用的资源情况，若是超过了该container所声明的资源量，则会kill掉该container所代表的进程。</p>
<p>&nbsp;&nbsp;NM还提供了一个简单的服务以管理它所在机器的本地目录。Applications可以继续访问本地目录即使那台机器上已经没有了属于它的container在运行。例如，Map-Reduce应用程序使用这个服务存储map output并且shuffle它们给相应的reduce task。</p>
<p>&nbsp;&nbsp;NM上还可以扩展自己的服务，yarn提供了一个yarn.nodemanager.aux-services的配置项，通过该配置，用户可以自定义一些服务，例如Map-Reduce的shuffle功能就是采用这种方式实现的。</p>
<p>NM在本地为每个运行着的application生成如下的目录结构：</p>
<p><img src="http://ww3.sinaimg.cn/mw690/63503acbjw1f5w44b7lxfj208u05dt9v.jpg"></p>
<p>Container目录下的目录结构如下： </p>
<p><img src="http://ww1.sinaimg.cn/mw690/63503acbjw1f5w44bjrc6j206m05jq3d.jpg"></p>
<p>&nbsp;&nbsp;在启动一个container的时候，NM就执行该container的default_container_executor.sh，该脚本内部会执行launch_container.sh。launch_container.sh会先设置一些环境变量，最后启动执行程序的命令。对于MapReduce而言，启动AM就执行org.apache.hadoop.mapreduce.v2.app.MRAppMaster；启动map/reduce task就执行org.apache.hadoop.mapred.YarnChild。 </p>
<p><strong>ApplicationMaster</strong></p>
<p>&nbsp;&nbsp;ApplicationMaster是一个框架特殊的库，对于Map-Reduce计算模型而言有它自己的ApplicationMaster实现，对于其他的想要运行在yarn上的计算模型而言，必须得实现针对该计算模型的ApplicationMaster用以向RM申请资源运行task，比如运行在yarn上的spark框架也有对应的ApplicationMaster实现，归根结底，yarn是一个资源管理的框架，并不是一个计算框架，要想在yarn上运行应用程序，还得有特定的计算框架的实现。</p>
<h4 id="工作流程"><a href="#工作流程" class="headerlink" title="工作流程"></a>工作流程</h4><p><img src="http://ww1.sinaimg.cn/mw690/63503acbjw1f5w48pm9byj20ui0kl0uy.jpg"></p>
<ol>
<li>JobClient向ResourceManager(AsM)申请提交一个job。</li>
<li>RM返回jobId和job提交路径。</li>
<li>JobClient提交job相关的文件。</li>
<li>向RM汇报提交完成。</li>
<li>RM将job写入Job Queue。</li>
<li>NodeManager(NM)向Job Queue领取任务。</li>
<li>ApplicationMaster(AM)启动,向RM进行注册。</li>
<li>RM向AM返回资源信息。</li>
<li>AM启动map。</li>
<li>当所有map任务完成后,AM启动reduce。 </li>
<li>AM监视运行着的task直到完成,当task失败时,申请新的container运行失败的task。</li>
<li>当每个map/reduce task完成后,AM运行MR OutputCommitter的cleanup 代码，进行一些收尾工作。</li>
<li>当所有的map/reduce完成后,AM运行OutputCommitter的必要的job commit或者abort APIs。</li>
<li>AM注销自己。</li>
</ol>
<h3 id="Shuffle过程"><a href="#Shuffle过程" class="headerlink" title="Shuffle过程"></a>Shuffle过程</h3><p><img src="http://ww3.sinaimg.cn/mw690/63503acbjw1f5w449wf5hj20mv0aodh4.jpg"></p>
<p><img src="http://ww4.sinaimg.cn/mw690/63503acbjw1f5w44an439j20k10g5mzg.jpg"></p>
<h4 id="Map"><a href="#Map" class="headerlink" title="Map"></a>Map</h4><ol>
<li><p>每个输入分片会让一个map任务来处理，默认情况下，以HDFS的一个块的大小（默认为64M）为一个分片，当然我们也可以设置块的大小。map输出的结果会暂且放在一个环形内存缓冲区中（该缓冲区的大小默认为100M，由io.sort.mb属性控制），当该缓冲区快要溢出时（默认为缓冲区大小的80%，由io.sort.spill.percent属性控制），会在本地文件系统中创建一个溢出文件，将该缓冲区中的数据写入这个文件。</p>
</li>
<li><p>在写入磁盘之前，线程首先根据reduce任务的数目将数据划分为相同数目的分区，也就是一个reduce任务对应一个分区的数据。这样做是为了避免有些reduce任务分配到大量数据，而有些reduce任务却分到很少数据，甚至没有分到数据的尴尬局面。其实分区就是对数据进行hash的过程。然后对每个分区中的数据进行排序，如果此时设置了Combiner，将排序后的结果进行Combia操作，这样做的目的是让尽可能少的数据写入到磁盘。</p>
</li>
<li><p>当map任务输出最后一个记录时，可能会有很多的溢出文件，这时需要将这些文件合并。合并的过程中会不断地进行排序和combia操作，目的有两个：</p>
<ul>
<li>尽量减少每次写入磁盘的数据量；</li>
<li>尽量减少下一复制阶段网络传输的数据量。最后合并成了一个已分区且已排序的文件。</li>
</ul>
<p>为了减少网络传输的数据量，这里可以将数据压缩，只要将mapred.compress.map.out设置为true就可以了。</p>
</li>
<li><p>将分区中的数据拷贝给相对应的reduce Task。有人可能会问：分区中的数据怎么知道它对应的reduce是哪个呢？其实map任务一直和其父TaskTracker保持联系，而TaskTracker又一直和JobTracker保持心跳。所以JobTracker中保存了整个集群中的宏观信息。只要reduce任务向JobTracker获取对应的map输出位置就ok了哦。</p>
</li>
</ol>
<h4 id="Reduce"><a href="#Reduce" class="headerlink" title="Reduce"></a>Reduce</h4><ol>
<li><p>Reduce会接收到不同map任务传来的数据，并且每个map传来的数据都是有序的。如果reduce端接受的数据量相当小，则直接存储在内存中（缓冲区大小由mapred.job.shuffle.input.buffer.percent属性控制，表示用作此用途的堆空间的百分比），如果数据量超过了该缓冲区大小的一定比例（由mapred.job.shuffle.merge.percent决定），则对数据合并后溢写到磁盘中。</p>
</li>
<li><p>随着溢写文件的增多，后台线程会将它们合并成一个更大的有序的文件，这样做是为了给后面的合并节省时间。其实不管在map端还是reduce端，MapReduce都是反复地执行排序，合并操作。</p>
</li>
<li><p>合并的过程中会产生许多的中间文件（写入磁盘了），但MapReduce会让写入磁盘的数据尽可能地少，并且最后一次合并的结果并没有写入磁盘，而是直接输入到reduce函数。</p>
</li>
</ol>
<h3 id="WordCount案例"><a href="#WordCount案例" class="headerlink" title="WordCount案例"></a>WordCount案例</h3><h4 id="Mapper"><a href="#Mapper" class="headerlink" title="Mapper"></a>Mapper</h4><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">MyWordCountMapper</span> <span class="keyword">extends</span> <span class="title">Mapper</span>&lt;<span class="title">LongWritable</span>, <span class="title">Text</span>, <span class="title">Text</span>, <span class="title">LongWritable</span>&gt; </span>&#123;</span><br><span class="line"></span><br><span class="line">	<span class="meta">@Override</span></span><br><span class="line">	<span class="function"><span class="keyword">protected</span> <span class="keyword">void</span> <span class="title">map</span><span class="params">(LongWritable key, Text value, Context context)</span> <span class="keyword">throws</span> IOException, InterruptedException </span>&#123;</span><br><span class="line">		<span class="comment">// 读取一行的value</span></span><br><span class="line">		String line = value.toString();</span><br><span class="line">		<span class="comment">// 按照规则切分</span></span><br><span class="line">		String[] words = line.split(<span class="string">&quot; &quot;</span>);</span><br><span class="line">		<span class="comment">// 按照&lt;单词,1&gt;的格式输出</span></span><br><span class="line">		<span class="keyword">for</span> (String word : words) &#123;</span><br><span class="line">			context.write(<span class="keyword">new</span> Text(word), <span class="keyword">new</span> LongWritable(<span class="number">1</span>));</span><br><span class="line">		&#125;</span><br><span class="line">	&#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<h4 id="Reducer"><a href="#Reducer" class="headerlink" title="Reducer"></a>Reducer</h4><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">MyWordCountReducer</span> <span class="keyword">extends</span> <span class="title">Reducer</span>&lt;<span class="title">Text</span>, <span class="title">LongWritable</span>, <span class="title">Text</span>, <span class="title">LongWritable</span>&gt; </span>&#123;</span><br><span class="line"></span><br><span class="line">	<span class="meta">@Override</span></span><br><span class="line">	<span class="function"><span class="keyword">protected</span> <span class="keyword">void</span> <span class="title">reduce</span><span class="params">(Text key, Iterable&lt;LongWritable&gt; values, Context context)</span></span></span><br><span class="line"><span class="function">			<span class="keyword">throws</span> IOException, InterruptedException </span>&#123;</span><br><span class="line">		<span class="comment">// 初始化计数器</span></span><br><span class="line">		<span class="keyword">long</span> count = <span class="number">0</span>;</span><br><span class="line">		<span class="comment">// 迭代values,累加计数器计算出总次数</span></span><br><span class="line">		<span class="keyword">for</span> (LongWritable value : values) &#123;</span><br><span class="line">			count += value.get();</span><br><span class="line">		&#125;</span><br><span class="line">		<span class="comment">// 输出&lt;单词,总次数&gt;</span></span><br><span class="line">		context.write(key, <span class="keyword">new</span> LongWritable(count));</span><br><span class="line">	&#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<h4 id="Main"><a href="#Main" class="headerlink" title="Main"></a>Main</h4><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">MyWordCountDriver</span> </span>&#123;</span><br><span class="line"></span><br><span class="line">	<span class="function"><span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">void</span> <span class="title">main</span><span class="params">(String[] args)</span></span></span><br><span class="line"><span class="function">	    <span class="keyword">throws</span> IOException, ClassNotFoundException, InterruptedException </span>&#123;</span><br><span class="line">		Configuration conf = <span class="keyword">new</span> Configuration();</span><br><span class="line">		<span class="comment">// 构造一个job对象</span></span><br><span class="line">		Job wordCountJob = Job.getInstance(conf);</span><br><span class="line"></span><br><span class="line">		<span class="comment">// 指定job用到的jar包位置,这里使用当前类</span></span><br><span class="line">		wordCountJob.setJarByClass(MyWordCountDriver.class);</span><br><span class="line"></span><br><span class="line">		<span class="comment">// 指定mapper</span></span><br><span class="line">		wordCountJob.setMapperClass(MyWordCountMapper.class);</span><br><span class="line">		<span class="comment">// 指定reducer</span></span><br><span class="line">		wordCountJob.setReducerClass(MyWordCountReducer.class);</span><br><span class="line"></span><br><span class="line">		<span class="comment">// 指定mapper输出key/value的类型</span></span><br><span class="line">		wordCountJob.setMapOutputKeyClass(Text.class);</span><br><span class="line">		wordCountJob.setMapOutputValueClass(LongWritable.class);</span><br><span class="line"></span><br><span class="line">		<span class="comment">// 指定reducer输出key/value的类型</span></span><br><span class="line">		wordCountJob.setOutputKeyClass(Text.class);</span><br><span class="line">		wordCountJob.setOutputValueClass(LongWritable.class);</span><br><span class="line"></span><br><span class="line">		<span class="comment">// 指定输入数据的路径</span></span><br><span class="line">		FileInputFormat.setInputPaths(wordCountJob, <span class="keyword">new</span> Path(args[<span class="number">0</span>]));</span><br><span class="line">		<span class="comment">// 指定输出结果的路径</span></span><br><span class="line">		FileOutputFormat.setOutputPath(wordCountJob, <span class="keyword">new</span> Path(args[<span class="number">1</span>]));</span><br><span class="line">		<span class="comment">// 通过yarn客户端进行提交,参数2为是否打印到控制台</span></span><br><span class="line">		wordCountJob.waitForCompletion(<span class="keyword">true</span>);</span><br><span class="line">	&#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<h4 id="启动MapReduce"><a href="#启动MapReduce" class="headerlink" title="启动MapReduce"></a>启动MapReduce</h4><p>&nbsp;&nbsp;<strong>方式1:</strong> 将程序打成jar包,上传到hadoop中执行。hadoop jar <jar> [mainClass] args…</p>
<p>&nbsp;&nbsp;<strong>方式2:</strong> 将程序打成jar包,在本地IDE上直接运行(需要代码指定jar)。</p>
<h3 id="自定义Sort"><a href="#自定义Sort" class="headerlink" title="自定义Sort"></a>自定义Sort</h3><h4 id="bean"><a href="#bean" class="headerlink" title="bean"></a>bean</h4><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br><span class="line">47</span><br><span class="line">48</span><br><span class="line">49</span><br><span class="line">50</span><br><span class="line">51</span><br><span class="line">52</span><br><span class="line">53</span><br><span class="line">54</span><br><span class="line">55</span><br><span class="line">56</span><br><span class="line">57</span><br><span class="line">58</span><br><span class="line">59</span><br><span class="line">60</span><br><span class="line">61</span><br><span class="line">62</span><br><span class="line">63</span><br><span class="line">64</span><br><span class="line">65</span><br><span class="line">66</span><br><span class="line">67</span><br><span class="line">68</span><br><span class="line">69</span><br><span class="line">70</span><br><span class="line">71</span><br><span class="line">72</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">FlowBean</span> <span class="keyword">implements</span> <span class="title">WritableComparable</span>&lt;<span class="title">FlowBean</span>&gt; </span>&#123;</span><br><span class="line"></span><br><span class="line">	<span class="keyword">private</span> Long upFlow;</span><br><span class="line">	<span class="keyword">private</span> Long downFlow;</span><br><span class="line">	<span class="keyword">private</span> Long sumFlow;</span><br><span class="line"></span><br><span class="line">	<span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">setAll</span><span class="params">(Long upFlow, Long downFlow)</span> </span>&#123;</span><br><span class="line">		<span class="keyword">this</span>.upFlow = upFlow;</span><br><span class="line">		<span class="keyword">this</span>.downFlow = downFlow;</span><br><span class="line">		<span class="keyword">this</span>.sumFlow = upFlow + downFlow;</span><br><span class="line">	&#125;</span><br><span class="line"></span><br><span class="line">	<span class="function"><span class="keyword">public</span> Long <span class="title">getUpFlow</span><span class="params">()</span> </span>&#123;</span><br><span class="line">		<span class="keyword">return</span> upFlow;</span><br><span class="line">	&#125;</span><br><span class="line"></span><br><span class="line">	<span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">setUpFlow</span><span class="params">(Long upFlow)</span> </span>&#123;</span><br><span class="line">		<span class="keyword">this</span>.upFlow = upFlow;</span><br><span class="line">	&#125;</span><br><span class="line"></span><br><span class="line">	<span class="function"><span class="keyword">public</span> Long <span class="title">getDownFlow</span><span class="params">()</span> </span>&#123;</span><br><span class="line">		<span class="keyword">return</span> downFlow;</span><br><span class="line">	&#125;</span><br><span class="line"></span><br><span class="line">	<span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">setDownFlow</span><span class="params">(Long downFlow)</span> </span>&#123;</span><br><span class="line">		<span class="keyword">this</span>.downFlow = downFlow;</span><br><span class="line">	&#125;</span><br><span class="line"></span><br><span class="line">	<span class="function"><span class="keyword">public</span> Long <span class="title">getSumFlow</span><span class="params">()</span> </span>&#123;</span><br><span class="line">		<span class="keyword">return</span> sumFlow;</span><br><span class="line">	&#125;</span><br><span class="line"></span><br><span class="line">	<span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">setSumFlow</span><span class="params">(Long sumFlow)</span> </span>&#123;</span><br><span class="line">		<span class="keyword">this</span>.sumFlow = sumFlow;</span><br><span class="line">	&#125;</span><br><span class="line"></span><br><span class="line">	<span class="meta">@Override</span></span><br><span class="line">	<span class="function"><span class="keyword">public</span> String <span class="title">toString</span><span class="params">()</span> </span>&#123;</span><br><span class="line">		<span class="keyword">return</span> <span class="string">&quot;FlowBean [upFlow=&quot;</span> + upFlow + <span class="string">&quot;, downFlow=&quot;</span> + downFlow</span><br><span class="line">				+ <span class="string">&quot;, sumFlow=&quot;</span> + sumFlow + <span class="string">&quot;]&quot;</span>;</span><br><span class="line">	&#125;</span><br><span class="line"></span><br><span class="line">	<span class="comment">/**</span></span><br><span class="line"><span class="comment">	 * 序列化</span></span><br><span class="line"><span class="comment">	 */</span></span><br><span class="line">	<span class="meta">@Override</span></span><br><span class="line">	<span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">write</span><span class="params">(DataOutput out)</span> <span class="keyword">throws</span> IOException </span>&#123;</span><br><span class="line">		out.writeLong(upFlow);</span><br><span class="line">		out.writeLong(downFlow);</span><br><span class="line">		out.writeLong(sumFlow);</span><br><span class="line">	&#125;</span><br><span class="line"></span><br><span class="line">	<span class="comment">/**</span></span><br><span class="line"><span class="comment">	 * 反序列化</span></span><br><span class="line"><span class="comment">	 */</span></span><br><span class="line">	<span class="meta">@Override</span></span><br><span class="line">	<span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">readFields</span><span class="params">(DataInput in)</span> <span class="keyword">throws</span> IOException </span>&#123;</span><br><span class="line">		upFlow = in.readLong();</span><br><span class="line">		downFlow = in.readLong();</span><br><span class="line">		sumFlow = in.readLong();</span><br><span class="line">	&#125;</span><br><span class="line"></span><br><span class="line">	<span class="comment">/**</span></span><br><span class="line"><span class="comment">	 * 降序排序 -1 大于 0 等于 1小于</span></span><br><span class="line"><span class="comment">	 */</span></span><br><span class="line">	<span class="meta">@Override</span></span><br><span class="line">	<span class="function"><span class="keyword">public</span> <span class="keyword">int</span> <span class="title">compareTo</span><span class="params">(FlowBean o)</span> </span>&#123;</span><br><span class="line">		<span class="comment">// 如果当前类的总和大于其他类的总和 则返回-1(大于) false 1(小于)</span></span><br><span class="line">		<span class="keyword">return</span> <span class="keyword">this</span>.sumFlow &gt; o.getSumFlow() ? -<span class="number">1</span> : <span class="number">1</span>;</span><br><span class="line">	&#125;</span><br><span class="line"></span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<h4 id="main"><a href="#main" class="headerlink" title="main"></a>main</h4><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br><span class="line">47</span><br><span class="line">48</span><br><span class="line">49</span><br><span class="line">50</span><br><span class="line">51</span><br><span class="line">52</span><br><span class="line">53</span><br><span class="line">54</span><br><span class="line">55</span><br><span class="line">56</span><br><span class="line">57</span><br><span class="line">58</span><br><span class="line">59</span><br><span class="line">60</span><br><span class="line">61</span><br><span class="line">62</span><br><span class="line">63</span><br><span class="line">64</span><br><span class="line">65</span><br><span class="line">66</span><br><span class="line">67</span><br><span class="line">68</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">FlowSummarySort</span> </span>&#123;</span><br><span class="line"></span><br><span class="line">	<span class="comment">/**</span></span><br><span class="line"><span class="comment">	 * 因为只有key才能进行排序,所以输出key为FlowBean</span></span><br><span class="line"><span class="comment">	 * </span></span><br><span class="line"><span class="comment">	 * <span class="doctag">@author</span> sylvanasp</span></span><br><span class="line"><span class="comment">	 * <span class="doctag">@version</span> 1.0</span></span><br><span class="line"><span class="comment">	 */</span></span><br><span class="line">	<span class="keyword">public</span> <span class="keyword">static</span> <span class="class"><span class="keyword">class</span> <span class="title">FlowSummarySortMapper</span></span></span><br><span class="line"><span class="class">			<span class="keyword">extends</span> <span class="title">Mapper</span>&lt;<span class="title">LongWritable</span>, <span class="title">Text</span>, <span class="title">FlowBean</span>, <span class="title">Text</span>&gt; </span>&#123;</span><br><span class="line"></span><br><span class="line">		<span class="meta">@Override</span></span><br><span class="line">		<span class="function"><span class="keyword">protected</span> <span class="keyword">void</span> <span class="title">map</span><span class="params">(LongWritable key, Text value, Context context)</span></span></span><br><span class="line"><span class="function">				<span class="keyword">throws</span> IOException, InterruptedException </span>&#123;</span><br><span class="line">			String line = value.toString();</span><br><span class="line">			String[] fields = StringUtils.split(line, <span class="string">&quot;\t&quot;</span>);</span><br><span class="line">			String phoneNum = fields[<span class="number">0</span>];</span><br><span class="line">			Long upFlow = Long.parseLong(fields[<span class="number">1</span>]);</span><br><span class="line">			Long downFlow = Long.parseLong(fields[<span class="number">2</span>]);</span><br><span class="line"></span><br><span class="line">			FlowBean flowBean = <span class="keyword">new</span> FlowBean();</span><br><span class="line">			flowBean.setAll(upFlow, downFlow);</span><br><span class="line">			context.write(flowBean, <span class="keyword">new</span> Text(phoneNum));</span><br><span class="line">		&#125;</span><br><span class="line"></span><br><span class="line">	&#125;</span><br><span class="line"></span><br><span class="line">	<span class="comment">/**</span></span><br><span class="line"><span class="comment">	 * 因为在mapper中已经完成了排序,所以reducer中需要将phoneNum重新设置为key</span></span><br><span class="line"><span class="comment">	 * </span></span><br><span class="line"><span class="comment">	 * <span class="doctag">@author</span> sylvanasp</span></span><br><span class="line"><span class="comment">	 * <span class="doctag">@version</span> 1.0</span></span><br><span class="line"><span class="comment">	 */</span></span><br><span class="line">	<span class="keyword">public</span> <span class="keyword">static</span> <span class="class"><span class="keyword">class</span> <span class="title">FlowSummarySortReducer</span></span></span><br><span class="line"><span class="class">			<span class="keyword">extends</span> <span class="title">Reducer</span>&lt;<span class="title">FlowBean</span>, <span class="title">Text</span>, <span class="title">Text</span>, <span class="title">FlowBean</span>&gt; </span>&#123;</span><br><span class="line"></span><br><span class="line">		<span class="meta">@Override</span></span><br><span class="line">		<span class="function"><span class="keyword">protected</span> <span class="keyword">void</span> <span class="title">reduce</span><span class="params">(FlowBean bean, Iterable&lt;Text&gt; phoneNum,</span></span></span><br><span class="line"><span class="function"><span class="params">				Context context)</span> <span class="keyword">throws</span> IOException, InterruptedException </span>&#123;</span><br><span class="line">			<span class="comment">// 因为每个bean都是完全独立的,所以Iterable中只有一个数据</span></span><br><span class="line">			<span class="keyword">for</span> (Text phoneNumKey : phoneNum) &#123;</span><br><span class="line">				context.write(phoneNumKey, bean);</span><br><span class="line">			&#125;</span><br><span class="line">		&#125;</span><br><span class="line"></span><br><span class="line">	&#125;</span><br><span class="line"></span><br><span class="line">	<span class="function"><span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">void</span> <span class="title">main</span><span class="params">(String[] args)</span> <span class="keyword">throws</span> Exception </span>&#123;</span><br><span class="line">		Configuration conf = <span class="keyword">new</span> Configuration();</span><br><span class="line">		Job job = Job.getInstance(conf);</span><br><span class="line">		job.setJarByClass(FlowSummarySort.class);</span><br><span class="line">		job.setMapperClass(FlowSummarySortMapper.class);</span><br><span class="line">		job.setReducerClass(FlowSummarySortReducer.class);</span><br><span class="line"></span><br><span class="line">		job.setMapOutputKeyClass(FlowBean.class);</span><br><span class="line">		job.setMapOutputValueClass(Text.class);</span><br><span class="line"></span><br><span class="line">		job.setOutputKeyClass(Text.class);</span><br><span class="line">		job.setOutputValueClass(FlowBean.class);</span><br><span class="line"></span><br><span class="line">		FileInputFormat.setInputPaths(job, <span class="keyword">new</span> Path(args[<span class="number">0</span>]));</span><br><span class="line">		FileOutputFormat.setOutputPath(job, <span class="keyword">new</span> Path(args[<span class="number">1</span>]));</span><br><span class="line"></span><br><span class="line">		<span class="keyword">int</span> result = job.waitForCompletion(<span class="keyword">true</span>) ? <span class="number">0</span> : <span class="number">1</span>;</span><br><span class="line">		System.exit(result);</span><br><span class="line">	&#125;</span><br><span class="line"></span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<h3 id="自定义Partition"><a href="#自定义Partition" class="headerlink" title="自定义Partition"></a>自定义Partition</h3><h4 id="partitioner"><a href="#partitioner" class="headerlink" title="partitioner"></a>partitioner</h4><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">MyPartitioner</span> <span class="keyword">extends</span> <span class="title">Partitioner</span>&lt;<span class="title">Text</span>, <span class="title">FlowBean</span>&gt; </span>&#123;</span><br><span class="line"></span><br><span class="line">	<span class="comment">// 使用map模拟数据库</span></span><br><span class="line">	<span class="keyword">private</span> <span class="keyword">static</span> HashMap&lt;String, Integer&gt; map = <span class="keyword">new</span> HashMap&lt;String, Integer&gt;();</span><br><span class="line"></span><br><span class="line">	<span class="comment">// 初始化分区规则</span></span><br><span class="line">	<span class="keyword">static</span> &#123;</span><br><span class="line">		map.put(<span class="string">&quot;136&quot;</span>, <span class="number">0</span>);</span><br><span class="line">		map.put(<span class="string">&quot;137&quot;</span>, <span class="number">1</span>);</span><br><span class="line">		map.put(<span class="string">&quot;138&quot;</span>, <span class="number">2</span>);</span><br><span class="line">		map.put(<span class="string">&quot;139&quot;</span>, <span class="number">3</span>);</span><br><span class="line">	&#125;</span><br><span class="line"></span><br><span class="line">	<span class="meta">@Override</span></span><br><span class="line">	<span class="function"><span class="keyword">public</span> <span class="keyword">int</span> <span class="title">getPartition</span><span class="params">(Text key, FlowBean value, <span class="keyword">int</span> numPartitions)</span> </span>&#123;</span><br><span class="line">		<span class="comment">// 获取手机号前3位</span></span><br><span class="line">		String phonePrefix = key.toString().substring(<span class="number">0</span>, <span class="number">3</span>);</span><br><span class="line">		<span class="comment">// 根据手机号前缀获得对应的分区编号</span></span><br><span class="line">		Integer partitionId = map.get(phonePrefix);</span><br><span class="line">		<span class="comment">// 如果手机号不在分区规则内,则分配到分区4。</span></span><br><span class="line">		<span class="keyword">return</span> partitionId == <span class="keyword">null</span> ? <span class="number">4</span> : partitionId;</span><br><span class="line">	&#125;</span><br><span class="line"></span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<h4 id="main-1"><a href="#main-1" class="headerlink" title="main"></a>main</h4><figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br><span class="line">22</span><br><span class="line">23</span><br><span class="line">24</span><br><span class="line">25</span><br><span class="line">26</span><br><span class="line">27</span><br><span class="line">28</span><br><span class="line">29</span><br><span class="line">30</span><br><span class="line">31</span><br><span class="line">32</span><br><span class="line">33</span><br><span class="line">34</span><br><span class="line">35</span><br><span class="line">36</span><br><span class="line">37</span><br><span class="line">38</span><br><span class="line">39</span><br><span class="line">40</span><br><span class="line">41</span><br><span class="line">42</span><br><span class="line">43</span><br><span class="line">44</span><br><span class="line">45</span><br><span class="line">46</span><br><span class="line">47</span><br><span class="line">48</span><br><span class="line">49</span><br><span class="line">50</span><br><span class="line">51</span><br><span class="line">52</span><br><span class="line">53</span><br><span class="line">54</span><br><span class="line">55</span><br><span class="line">56</span><br><span class="line">57</span><br><span class="line">58</span><br><span class="line">59</span><br><span class="line">60</span><br><span class="line">61</span><br><span class="line">62</span><br><span class="line">63</span><br><span class="line">64</span><br><span class="line">65</span><br><span class="line">66</span><br></pre></td><td class="code"><pre><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">FlowSummaryPartition</span> </span>&#123;</span><br><span class="line"></span><br><span class="line">	<span class="keyword">public</span> <span class="keyword">static</span> <span class="class"><span class="keyword">class</span> <span class="title">FlowSummaryPartitionMapper</span></span></span><br><span class="line"><span class="class">			<span class="keyword">extends</span> <span class="title">Mapper</span>&lt;<span class="title">LongWritable</span>, <span class="title">Text</span>, <span class="title">Text</span>, <span class="title">FlowBean</span>&gt; </span>&#123;</span><br><span class="line"></span><br><span class="line">		<span class="meta">@Override</span></span><br><span class="line">		<span class="function"><span class="keyword">protected</span> <span class="keyword">void</span> <span class="title">map</span><span class="params">(LongWritable key, Text value, Context context)</span></span></span><br><span class="line"><span class="function">				<span class="keyword">throws</span> IOException, InterruptedException </span>&#123;</span><br><span class="line">			String line = value.toString();</span><br><span class="line">			String[] fields = StringUtils.split(line, <span class="string">&quot;\t&quot;</span>);</span><br><span class="line"></span><br><span class="line">			String phoneNum = fields[<span class="number">1</span>];</span><br><span class="line">			Long upFlow = Long.parseLong(fields[fields.length - <span class="number">3</span>]);</span><br><span class="line">			Long downFlow = Long.parseLong(fields[fields.length - <span class="number">2</span>]);</span><br><span class="line"></span><br><span class="line">			FlowBean flowBean = <span class="keyword">new</span> FlowBean();</span><br><span class="line">			flowBean.setAll(upFlow, downFlow);</span><br><span class="line">			context.write(<span class="keyword">new</span> Text(phoneNum), flowBean);</span><br><span class="line">		&#125;</span><br><span class="line"></span><br><span class="line">	&#125;</span><br><span class="line"></span><br><span class="line">	<span class="keyword">public</span> <span class="keyword">static</span> <span class="class"><span class="keyword">class</span> <span class="title">FlowSummaryPartitionReducer</span></span></span><br><span class="line"><span class="class">			<span class="keyword">extends</span> <span class="title">Reducer</span>&lt;<span class="title">Text</span>, <span class="title">FlowBean</span>, <span class="title">Text</span>, <span class="title">FlowBean</span>&gt; </span>&#123;</span><br><span class="line"></span><br><span class="line">		<span class="meta">@Override</span></span><br><span class="line">		<span class="function"><span class="keyword">protected</span> <span class="keyword">void</span> <span class="title">reduce</span><span class="params">(Text key, Iterable&lt;FlowBean&gt; beans,</span></span></span><br><span class="line"><span class="function"><span class="params">				Context context)</span> <span class="keyword">throws</span> IOException, InterruptedException </span>&#123;</span><br><span class="line">			<span class="keyword">long</span> upSum = <span class="number">0</span>;</span><br><span class="line">			<span class="keyword">long</span> downSum = <span class="number">0</span>;</span><br><span class="line"></span><br><span class="line">			<span class="keyword">for</span> (FlowBean bean : beans) &#123;</span><br><span class="line">				upSum += bean.getUpFlow();</span><br><span class="line">				downSum += bean.getDownFlow();</span><br><span class="line">			&#125;</span><br><span class="line">			FlowBean flowBean = <span class="keyword">new</span> FlowBean();</span><br><span class="line">			flowBean.setAll(upSum, downSum);</span><br><span class="line">			context.write(key, flowBean);</span><br><span class="line">		&#125;</span><br><span class="line"></span><br><span class="line">	&#125;</span><br><span class="line"></span><br><span class="line">	<span class="function"><span class="keyword">public</span> <span class="keyword">static</span> <span class="keyword">void</span> <span class="title">main</span><span class="params">(String[] args)</span> <span class="keyword">throws</span> Exception </span>&#123;</span><br><span class="line">		Configuration conf = <span class="keyword">new</span> Configuration();</span><br><span class="line">		Job job = Job.getInstance(conf);</span><br><span class="line"></span><br><span class="line">		job.setJarByClass(FlowSummaryPartition.class);</span><br><span class="line">		job.setMapperClass(FlowSummaryPartitionMapper.class);</span><br><span class="line">		job.setReducerClass(FlowSummaryPartitionReducer.class);</span><br><span class="line"></span><br><span class="line">		job.setOutputKeyClass(Text.class);</span><br><span class="line">		job.setOutputValueClass(FlowBean.class);</span><br><span class="line"></span><br><span class="line">		<span class="comment">// 设置分区器</span></span><br><span class="line">		job.setPartitionerClass(MyPartitioner.class);</span><br><span class="line">		<span class="comment">// 设置Reducer Task 实例数量 (与分区数一致)</span></span><br><span class="line">		job.setNumReduceTasks(<span class="number">5</span>);</span><br><span class="line"></span><br><span class="line">		FileInputFormat.setInputPaths(job, <span class="keyword">new</span> Path(args[<span class="number">0</span>]));</span><br><span class="line">		FileOutputFormat.setOutputPath(job, <span class="keyword">new</span> Path(args[<span class="number">1</span>]));</span><br><span class="line"></span><br><span class="line">		<span class="keyword">int</span> result = job.waitForCompletion(<span class="keyword">true</span>) ? <span class="number">0</span> : <span class="number">1</span>;</span><br><span class="line">		System.exit(result);</span><br><span class="line">	&#125;</span><br><span class="line"></span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>
<h3 id="END"><a href="#END" class="headerlink" title="END"></a>END</h3><blockquote>
<p>部分资料来源于<a target="_blank" rel="noopener" href="http://blog.sina.com.cn/s/blog_829a682d0101lc9d.html&amp;http://weixiaolu.iteye.com/blog/1474172">http://blog.sina.com.cn/s/blog_829a682d0101lc9d.html&amp;http://weixiaolu.iteye.com/blog/1474172</a></p>
</blockquote>

    </div>

    
    
    

    <footer class="post-footer">
          <div class="post-tags">
              <a href="/yuwanzi.io/tags/Hadoop/" rel="tag"># Hadoop</a>
              <a href="/yuwanzi.io/tags/%E5%A4%A7%E6%95%B0%E6%8D%AE/" rel="tag"># 大数据</a>
          </div>

        

          <div class="post-nav">
            <div class="post-nav-item">
                <a href="/yuwanzi.io/2016/07/12/2016-07-12-Hadoop-HDFS/" rel="prev" title="Hadoop学习笔记(1)-HDFS">
                  <i class="fa fa-chevron-left"></i> Hadoop学习笔记(1)-HDFS
                </a>
            </div>
            <div class="post-nav-item">
                <a href="/yuwanzi.io/2016/07/15/2016-07-15-Hadoop03-HA/" rel="next" title="Hadoop学习笔记(3)-HA高可用集群搭建">
                  Hadoop学习笔记(3)-HA高可用集群搭建 <i class="fa fa-chevron-right"></i>
                </a>
            </div>
          </div>
    </footer>
  </article>
</div>







<script>
  window.addEventListener('tabs:register', () => {
    let { activeClass } = CONFIG.comments;
    if (CONFIG.comments.storage) {
      activeClass = localStorage.getItem('comments_active') || activeClass;
    }
    if (activeClass) {
      const activeTab = document.querySelector(`a[href="#comment-${activeClass}"]`);
      if (activeTab) {
        activeTab.click();
      }
    }
  });
  if (CONFIG.comments.storage) {
    window.addEventListener('tabs:click', event => {
      if (!event.target.matches('.tabs-comment .tab-content .tab-pane')) return;
      const commentClass = event.target.classList[1];
      localStorage.setItem('comments_active', commentClass);
    });
  }
</script>
</div>
  </main>

  <footer class="footer">
    <div class="footer-inner">


<div class="copyright">
  &copy; 
  <span itemprop="copyrightYear">2021</span>
  <span class="with-love">
    <i class="fa fa-heart"></i>
  </span>
  <span class="author" itemprop="copyrightHolder">玉丸子</span>
</div>
  <div class="powered-by">Erstellt mit  <a href="https://hexo.io/" class="theme-link" rel="noopener" target="_blank">Hexo</a> & <a href="https://theme-next.js.org/muse/" class="theme-link" rel="noopener" target="_blank">NexT.Muse</a>
  </div>

    </div>
  </footer>

  
  <script src="//cdn.jsdelivr.net/npm/animejs@3.2.1/lib/anime.min.js"></script>
<script src="/yuwanzi.io/js/utils.js"></script><script src="/yuwanzi.io/js/motion.js"></script><script src="/yuwanzi.io/js/schemes/muse.js"></script><script src="/yuwanzi.io/js/next-boot.js"></script>

  






  





</body>
</html>
